package com.spark.core.persist

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object CacheAndPersist {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("cacheAndPersist")
    conf.setMaster("local[1]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("error")

    var lines = sc.textFile("data/words")

    /**
     * 一个application里面有多个job，多个job用到了同一个RDD，把这个RDD进行持久化
     */
    // 数据缓存到内存
    lines.cache()
    // persist(StorageLevel.MEMORY_ONLY)

    // 可以手动指定缓存级别
    // lines.persist(StorageLevel.MEMORY_AND_DISK)

    // 记录开始执行时间
    val startTime1 = System.currentTimeMillis()
    // 第一次数据来源是磁盘
    val count1 = lines.count()
    val endTime1 = System.currentTimeMillis()
    println("count1 = " + count1 + ",time = " + (endTime1 - startTime1) + "ms")

    val startTime2 = System.currentTimeMillis()
    // 第二次数据来源是内存
    val count2 = lines.count()
    val endTime2 = System.currentTimeMillis()
    println("count2 = " + count2 + ",time = " + (endTime2 - startTime2) + "ms")

    sc.stop()
  }
}
